/**************************************************************************
 //* W Corp. 
 //* W Profile Services                                             
 //* Package:  by.doubleyou.monitor.peer
 //* (C) Copyright W Corp.  2007
 /**************************************************************************
 *
 * Copyright (c) 2007, W Corporation
 * All rights reserved.
 * 
 * Version: 1.0
 * Date:  3 Dec, 2007, Author: Alexander J. Perez Tchernov
 * 
 * MODIFICATION HISTORY:
 * Defect/Feature  	Description
 * ---
 */
package by.doubleyou.monitor.peer;

import java.sql.SQLException;
import java.util.Arrays;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.logging.Level;

import by.doubleyou.core.gms.GMSClientService;
import by.doubleyou.monitor.db.LoggingStorage;

import com.sun.enterprise.ee.cms.core.MessageSignal;
import com.sun.enterprise.ee.cms.core.SignalAcquireException;
import com.sun.enterprise.ee.cms.core.SignalReleaseException;

/**
 * 
 * @author Alexander J. Perez Tchernov
 *
 */
public class StorageService extends GMSClientService{
	
	// Executor's threads
	//TODO config
	private int exThreads = 10;

	// Future executor
	private Executor exec = Executors.newFixedThreadPool(exThreads);
	
	private LoggingStorage storage = new LoggingStorage();


	/**
	 * Constructor
	 * 
	 * @param serviceName
	 * @param memberToken
	 * @param sendMessages
	 */
	public StorageService(final String serviceName, final String memberToken,
			final boolean sendMessages) {
		super(serviceName, memberToken, sendMessages);
	}
	
	@Override
	protected void processMessage( MessageSignal signal) {		
		super.processMessage(signal);
		try {
			try {
				signal.acquire();
				final MessageSignal s = signal;
				// executor 
				// berkeley like persistance cache (Tugela Cache) 
				// bulk loading 
				// sharding
				getLogger().log(Level.INFO, "[" + getServiceName() + ":" + getMemberToken() + "] " + new String(signal.getMessage()));
				final String  string = new String(s.getMessage());
				exec.execute(new Runnable() {
					public void run() {
						// TODO automata bricks
						String[] arrayInfo = string.split("\\s");
						System.out.println(Arrays.toString(arrayInfo));
						assert arrayInfo.length == 3 : "Malformed state";
						
						// message formeat:  ip time id
						try {
							storage.insertData( arrayInfo[0], arrayInfo[1], arrayInfo[2]);
						} catch (SQLException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
				});

			} finally {
				signal.release();
			}
		} catch (SignalAcquireException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (SignalReleaseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	@Override
	public void run() {
		super.run();
		try {
			Thread.currentThread().join();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} 
	}
	
	@Override
	public void stopClient() {
		super.stopClient();
		//TODO detach from db
	}

}
